package com.afdor.rws.event.driven.channel;

import com.afdor.rws.annotation.extension.Extension;
import com.afdor.rws.core.extension.ExtensionLoader;
import com.afdor.rws.core.strategy.IdGeneratorStrategy;
import com.afdor.rws.event.driven.Serialization;
import com.afdor.rws.event.driven.channel.consumer.EventConsumer;
import com.afdor.rws.event.driven.channel.consumer.handler.EventHandler;
import com.afdor.rws.event.driven.channel.consumer.handler.EventRetryStrategy;
import com.afdor.rws.event.driven.channel.producer.EventProducer;
import com.afdor.rws.event.driven.channel.transaction.ChannelTransactionItem;
import com.afdor.rws.event.driven.channel.transaction.impl.DefaultChannelTransactionItemImpl;
import com.afdor.rws.event.driven.config.EventChannelConfig;
import com.afdor.rws.event.driven.config.EventConsumerConfig;
import com.afdor.rws.event.driven.config.EventProducerConfig;
import com.afdor.rws.event.driven.entities.Event;
import com.afdor.rws.event.driven.entities.holder.CallbackEvent;
import com.afdor.rws.event.driven.entities.holder.ResultEvent;
import lombok.Getter;
import lombok.NoArgsConstructor;
import lombok.NonNull;

import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicMarkableReference;

/**
 * 事件传递执行通道默认实现
 *
 * @author 悭梵
 * @date Created in 2018-08-15 9:31
 */
@NoArgsConstructor
@Extension(value = "default", order = Extension.NORMAL_ORDER - 100, interfaceClass = ChannelProvider.class)
public class DefaultChannelProvider implements ChannelProvider {
    /**
     * 初始化状态
     */
    private volatile boolean init = Boolean.FALSE;
    /**
     * 配置信息
     */
    @NonNull
    private EventChannelConfig config;
    /**
     * 对应事件源
     */
    @Getter
    @NonNull
    private String eventSourceName;
    /**
     * 序列化方式
     */
    @Getter
    @NonNull
    private Serialization serialization;

    /**
     * 通道上的小芬
     */
    @Getter
    private final Map<String, Queue<EventConsumer>> eventConsumerMap = new ConcurrentHashMap<String, Queue<EventConsumer>>();

    /**
     * 生产者, 一个执行通道只有一个生产者
     */
    @Getter
    private AtomicMarkableReference<EventProducer> eventProducerReference = new AtomicMarkableReference<>(null, false);

    /**
     * 事物提交队列管理器
     */
    private ConcurrentMap<String, ChannelTransactionItem> eventChannelTransactionMap = new ConcurrentHashMap<String, ChannelTransactionItem>(64);

    @Override
    public <T extends ChannelProvider> T init() {
        this.init = Boolean.TRUE;
        return (T) this;
    }

    @Override
    public void destroy() {
        /**
         * 未提交事物全部回滚
         */
        eventChannelTransactionMap
                .entrySet()
                .parallelStream()
                .forEach(entry -> entry.getValue().rollback());
        /**
         * 生成通道关闭
         */
        getEventProducer().destroy();
        /**
         * 消费通道关闭
         */
        eventConsumerMap
                .entrySet()
                .parallelStream()
                .flatMap(entry -> entry.getValue().stream())
                .forEach(item -> item.destroy());
    }

    @Override
    public <C extends EventChannelConfig> C getConfig() {
        return (C) this.config;
    }

    @Override
    public <T extends ChannelProvider, C extends EventChannelConfig> T setConfig(@NonNull C config) {
        this.config = config;
        return (T) this;
    }

    @Override
    public <T extends ChannelProvider> T setEventSourceName(@NonNull String eventSourceName) {
        this.eventSourceName = eventSourceName;
        return (T) this;
    }

    @Override
    public <T extends ChannelProvider> T setSerialization(@NonNull Serialization serialization) {
        this.serialization = serialization;
        return (T) this;
    }

    @Override
    public <T extends ChannelProvider> T subscribe(@NonNull EventConsumerConfig eventConsumerConfig, @NonNull EventHandler eventHandler) {
        return subscribe(eventConsumerConfig, eventHandler, null);
    }

    @Override
    public <T extends ChannelProvider> T subscribe(@NonNull EventConsumerConfig eventConsumerConfig, @NonNull EventHandler eventHandler, EventRetryStrategy eventRetryStrategy) {
        Queue<EventConsumer> eventConsumerList = eventConsumerMap.get(getEventSourceName());
        if (eventConsumerList == null) {
            /* 因生产者、消费者可能都是多线程实现，故采用多线程安全队列 */
            eventConsumerMap.putIfAbsent(eventSourceName, new ConcurrentLinkedQueue<EventConsumer>());
            eventConsumerList = eventConsumerMap.get(eventSourceName);
        }

        EventConsumer eventConsumer = createEventConsumer(eventConsumerConfig, eventHandler, eventRetryStrategy);
        eventConsumerList.add(eventConsumer);
        eventConsumer.init();
        eventConsumer.start();
        return (T) this;
    }

    @Override
    public <V> ResultEvent<V> sendEvent(@NonNull Event event) {
        EventProducer eventProducer = getEventProducer();
        return eventProducer.sendEvent(event);
    }

    @Override
    public void sendAsyncEvent(@NonNull Event event) {
        EventProducer eventProducer = getEventProducer();
        eventProducer.sendAsyncEvent(event);
    }

    @Override
    public <V> void asyncSendEvent(@NonNull CallbackEvent callbackEvent) {
        EventProducer eventProducer = getEventProducer();
        eventProducer.asyncSendEvent(callbackEvent);
    }

    @Override
    public void batchSendAsyncEvent(@NonNull List<Event> eventList) {
        EventProducer eventProducer = getEventProducer();
        eventProducer.batchSendAsyncEvent(eventList);
    }

    /**
     * 获取事件对应的生产者<br/>
     * 一个Channel通道之能有一个生产者
     *
     * @return
     */
    protected EventProducer getEventProducer() {
        return getEventProducer(new EventProducerConfig(getConfig().getEventConfig()));
    }

    @NonNull
    @Override
    public ChannelTransactionItem newChannelTransaction() {
        ChannelTransactionItem channelTransactionItem = createChannelTransactionItem();
        if (this.eventChannelTransactionMap.containsKey(channelTransactionItem.getChannelTransactionId())) {
            return newChannelTransaction();
        } else {
            this.eventChannelTransactionMap.putIfAbsent(channelTransactionItem.getChannelTransactionId(), channelTransactionItem);
        }
        return channelTransactionItem;
    }

    @Override
    public void addEvent(@NonNull String channelTransactionId, @NonNull Event event) {
        ChannelTransactionItem channelTransactionItem = getChannelTransaction(channelTransactionId);
        channelTransactionItem.addEvent(event);
    }

    @Override
    public boolean commit(@NonNull String channelTransactionId) {
        try {
            ChannelTransactionItem channelTransactionItem = getChannelTransaction(channelTransactionId);
            return channelTransactionItem.commit();
        } finally {
            eventChannelTransactionMap.remove(channelTransactionId);
        }
    }

    @Override
    public void rollback(@NonNull String channelTransactionId) {
        try {
            ChannelTransactionItem channelTransactionItem = getChannelTransaction(channelTransactionId);
            channelTransactionItem.rollback();
        } finally {
            eventChannelTransactionMap.remove(channelTransactionId);
        }
    }

    /**
     * 获取执行通道事务管理器信息
     *
     * @param channelTransactionId
     * @return
     */
    protected ChannelTransactionItem getChannelTransaction(@NonNull String channelTransactionId) {
        return eventChannelTransactionMap.get(channelTransactionId);
    }

    /**
     * 根据主事物ID创建子事物管理器
     *
     * @return
     */
    protected ChannelTransactionItem createChannelTransactionItem() {
        return new DefaultChannelTransactionItemImpl(
                String.valueOf(ExtensionLoader
                        .getInstance(IdGeneratorStrategy.class)
                        .getExtensionOrDefault(getConfig().getTransactionIdStrategyName())
                        .generate()), this);
    }

    /**
     * 由通道具体实现方法来实现具体的消费者创建<br/>
     * 一个Channel通道可以有多个消费者，但只能有一个生产者
     *
     * @param eventHandler
     * @param eventRetryStrategy
     * @return
     */
    protected EventConsumer createEventConsumer(@NonNull EventConsumerConfig eventConsumerConfig, @NonNull EventHandler eventHandler, EventRetryStrategy eventRetryStrategy) {
        /* 构建并设置消费者对象 */
        return ExtensionLoader
                .getInstance(EventConsumer.class)
                .newExtension(getConfig().getEventConsumerStrategyName())
                .setEventHandler(eventHandler)
                .setEventRetryStrategy(eventRetryStrategy)
                .setSerialization(getSerialization())
                .setConfig(eventConsumerConfig)
                .setConsumerName(eventConsumerConfig.getConsumerName())
                .setEventSourceName(getEventSourceName());
    }

    /**
     * 获取事件对应的生产者<br/>
     * 一个Channel通道之能有一个生产者
     *
     * @return
     */
    protected EventProducer getEventProducer(EventProducerConfig eventProducerConfig) {
        if (!eventProducerReference.isMarked()) {

            EventProducer eventProducer = ExtensionLoader
                    .getInstance(EventProducer.class)
                    .newExtension(getConfig().getEventProducerStrategyName())
                    .setEventSourceName(getEventSourceName())
                    .setSerialization(getSerialization())
                    .setConfig(eventProducerConfig)
                    .init();

            eventProducerReference.compareAndSet(null, eventProducer, false, true);
        }
        return eventProducerReference.getReference();
    }

}
